package com.tom.architect.party3.day31;

final class ObservableSchedulers<T> extends Observable<T> {
    // 保存
    final Observable<T> source;
    final Schedulers schedulers;

    public ObservableSchedulers(Observable<T> source, Schedulers schedulers) {
        this.source = source;
        this.schedulers = schedulers;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        // 传入runnable对象
        schedulers.scheduleDirect(new SchedulerTask(observer));
    }

    private class SchedulerTask implements Runnable {
        final Observer<T> observer;
        public SchedulerTask(Observer<T> observer) {
            this.observer = observer;
        }
        @Override
        public void run() {
            // 线程池，最终会执行的Runnable，执行下面这行代码。会执行上游的 subscribe()方法。
            // 这个run方法在子线程中。调用上游的subscribe方法。不断地往上面调用。先往源头调用，再不断地调用onNext(map中的Function apply()方法)返回。
            source.subscribe(observer);
        }
    }
}
